package cas.ihep.hadoop.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

public class RedirectReader extends InputStreamReader {
    public FileSystem fs;
    public OutputStreamWriter hwriter;
    public boolean createdfs;

    public RedirectReader(String ifile,String outputhdfsFile,FileSystem fs_) throws IOException {
        super(new FileInputStream(ifile));
        createdfs=fs_==null;
        fs=createdfs?FileSystem.get(new Configuration()):fs_;
        hwriter=new OutputStreamWriter(fs.create(new Path(outputhdfsFile),true));
    }

    public void close()throws IOException{
        hwriter.close();
        super.close();
        if (createdfs){
            fs.close();
        }
    }

    public int read()throws IOException{
        int ch=super.read();
        hwriter.write(ch);
        return ch;
    }

    public int read(char buf[],int off,int len)throws IOException{
        int ret=super.read(buf,off,len);
        if(ret>0) {
            hwriter.write(buf, off, ret);
        }
        return ret;
    }

    public int read(char buf[])throws IOException{
        return this.read(buf,0,buf.length);
    }
}
